package com.gzyj.flink.gps.map.roadlevel;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.*;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper;
import com.gzyj.flink.gps.mapper.mysql.MonYejianxiansuSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.sink.gp.BaobiaoAlarmsummaryRecordSink;
import com.gzyj.flink.gps.sink.gp.BaobiaoAlarmsummarySink;
import com.gzyj.flink.gps.utils.IntUtils;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummaryRecord;
import com.gzyj.flink.pojo.mysql.set.MonChaosuSet;
import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet;
import com.gzyj.flink.pojo.mysql.set.MonYejianxiansuSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

public class RoadLevelLimittedFlow {

    private SingleOutputStreamOperator<Tuple3<GPSPosition, RoadLevelDataResult, MonChaosuSet>> roadlevelalarmStream;

    public RoadLevelLimittedFlow(SingleOutputStreamOperator<Tuple3<GPSPosition, RoadLevelDataResult, MonChaosuSet>> roadlevelalarmStream) {
        this.roadlevelalarmStream = roadlevelalarmStream;
    }

    public void process() {
        SingleOutputStreamOperator<RoadLevelResult> roadLevelResultSingleOutputStream = this.roadlevelalarmStream.map(new RichMapFunction<Tuple3<GPSPosition, RoadLevelDataResult, MonChaosuSet>, RoadLevelResult>() {
            LettuceConnectionFactory factory;
            SqlSessionFactory sqlSessionFactory;

            private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
                RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
                redisTemplate.setConnectionFactory(redisSessionFactory);
                redisTemplate.setKeySerializer(new StringRedisSerializer());
                redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.setHashKeySerializer(new StringRedisSerializer());
                redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.afterPropertiesSet();
                return redisTemplate;
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                factory = RedisSessionFactory.getRedisFactory();
                sqlSessionFactory = MySQLMybatisSessionFactory.getSqlSessionFactory();
            }

            @Override
            public void close() throws Exception {
                super.close();
            }

            private RoadLevelLimitedData TryGetData_RLLD(String vehicleId, RedisTemplate<String, Object> clusterRedisTemplate) {
                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_RoadLevelLimited + vehicleId);
                if (obj != null) {
                    return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), RoadLevelLimitedData.class);
                } else {
                    return new RoadLevelLimitedData();
                }
            }

            private MonYanzhongweiguiSet getVehicleForSeverityAlarmParamsDic(String vehiId, MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper) {
                return monYanzhongweiguiSetMapper.selectByVehId(vehiId);
            }

            public Pair<Boolean, MonYanzhongweiguiSet> GetSeverityRuleSetting(GPSPosition vehicle, MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
                MonYanzhongweiguiSet setting = getVehicleForSeverityAlarmParamsDic(vehicle.getVehId(), monYanzhongweiguiSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {

                    List<TsAlarmRule> list = tsAlarmRuleMapper.selectByRuleTypeYanzhong(vehicle.getShiyongxingzhi());

                    TsAlarmRule rule = list.size() > 0 ? list.get(0) : null;

                    if (rule == null || rule.getRuleFlag() == 0) {
                        return new Pair<>(false, setting);
                    } else {
                        setting = getVehicleForSeverityAlarmParamsDic(rule.getRuleNo(), monYanzhongweiguiSetMapper);
                        if (setting != null) {
                            return new Pair<>(true, setting);
                        } else {
                            return new Pair<>(false, setting);
                        }
                    }
                }
            }

            private Pair<Integer, Integer> CalcMaxSpeed(GPSPosition gpsPosition, Pair<Boolean, MonYanzhongweiguiSet> p, int limited) {
                int ruleType;
                if (p.getKey()) {
                    MonYanzhongweiguiSet severityAlarmSetting = p.getValue();
                    ruleType = severityAlarmSetting.getDingzhiguize();
                    if (severityAlarmSetting.getDingzhiguize() == 3) {
                        if (limited <= 40) {
                            return new Pair<>((int) (limited * 1.2), ruleType);

                        } else if (limited <= 50) {
                            return new Pair<>((int) (limited * 1.1), ruleType);
                        } else if (limited <= 60) {
                            return new Pair<>((int) (limited * 1.2), ruleType);
                        } else if (limited <= 70) {
                            return new Pair<>((int) (limited * 1.2), ruleType);
                        } else if (limited <= 80) {
                            return new Pair<>((int) (limited * 1.1), ruleType);
                        } else if (limited <= 90) {
                            return new Pair<>((int) (limited * 1.05), ruleType);
                        } else if (limited <= 100) {
                            return new Pair<>((int) (limited * 1.05), ruleType);
                        } else if (limited <= 110) {
                            return new Pair<>((int) (limited * 1.05), ruleType);
                        } else if (limited <= 120) {
                            return new Pair<>((int) (limited * 1.05), ruleType);
                        } else {
                            return new Pair<>(limited, ruleType);
                        }
                    }
                }
                ruleType = -1;
                return new Pair<>(limited, ruleType);
            }

            private MonYejianxiansuSet getVehicleForNightLimitedParamsDic(String vehiId, MonYejianxiansuSetMapper monYejianxiansuSetMapper) {
                return monYejianxiansuSetMapper.selectByVehID(vehiId);
            }

            private Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
                if (obj == null) {
                    return new Driverlogininfo();
                } else {
                    return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
                }

            }

            private Kafka_VehicleRoadLevelAlarm SendAlarmToKafka(GPSPosition gpsPosition, RoadLevelLimitedData data, boolean isAlarm, Driverlogininfo driverLoginData) {

                Kafka_VehicleRoadLevelAlarm kafkaVehicleRoadLevelAlarm = new Kafka_VehicleRoadLevelAlarm();
                kafkaVehicleRoadLevelAlarm.setVehId(gpsPosition.getVehId());
                kafkaVehicleRoadLevelAlarm.setDeptId(gpsPosition.getDeptId());
                kafkaVehicleRoadLevelAlarm.setVehNo(gpsPosition.getCheliangpaizhao());
                kafkaVehicleRoadLevelAlarm.setPlateColor(gpsPosition.getChepaiyanse());
                kafkaVehicleRoadLevelAlarm.setAlarmFlag(isAlarm ? 1 : 0);
                kafkaVehicleRoadLevelAlarm.setTime(TimeUtils.date2Str(data.getAlarmEndTime()));
                kafkaVehicleRoadLevelAlarm.setNote("超速报警(道路)");
                kafkaVehicleRoadLevelAlarm.setLongitude(data.getAlarmEndLongitude());
                kafkaVehicleRoadLevelAlarm.setLatitude(data.getAlarmEndLatitude());
                kafkaVehicleRoadLevelAlarm.setLastlongitude(data.getAlarmEndLongitude());
                kafkaVehicleRoadLevelAlarm.setLastlatitude(data.getAlarmEndLatitude());
                kafkaVehicleRoadLevelAlarm.setVelocity(gpsPosition.getSpeedExt());
                kafkaVehicleRoadLevelAlarm.setAngle(0);
                kafkaVehicleRoadLevelAlarm.setLocal(1);
                kafkaVehicleRoadLevelAlarm.setLimited(0);
                kafkaVehicleRoadLevelAlarm.setAlarmBeginTime(data.getAlarmBeginTime());
                kafkaVehicleRoadLevelAlarm.setAlarmEndTime(data.getAlarmEndTime());
                kafkaVehicleRoadLevelAlarm.setAlarmTimeSpan(Long.valueOf((data.getAlarmEndTime().getTime() - data.getAlarmBeginTime().getTime()) / 1000).intValue());
                kafkaVehicleRoadLevelAlarm.setMaxSpeed(data.getAlarmMaxSpeed());
                kafkaVehicleRoadLevelAlarm.setDistance(data.getAlarmDistance() * 1000);
                kafkaVehicleRoadLevelAlarm.setRoadLevel(data.getAlarmRoadLevel());
                kafkaVehicleRoadLevelAlarm.setRoadLimited(data.getAlarmRoadMaxSpeed());
                kafkaVehicleRoadLevelAlarm.setRoadName(data.getAlarmRoadName());
                kafkaVehicleRoadLevelAlarm.setAlarmGuid(data.getSingleGuid());
                kafkaVehicleRoadLevelAlarm.setAlarmLevel(data.getAlarmLevel());
                kafkaVehicleRoadLevelAlarm.setDriver(driverLoginData.getDriverName());
                kafkaVehicleRoadLevelAlarm.setDriverNo(driverLoginData.getDriverNo());
                kafkaVehicleRoadLevelAlarm.setAlarmSetting(data.getAlarmSetting());
                kafkaVehicleRoadLevelAlarm.setQualificationCode(driverLoginData.getDriverNo());
                kafkaVehicleRoadLevelAlarm.setKafkaFlag("Kafka_RoadLevelAlarm");

                return kafkaVehicleRoadLevelAlarm;


            }

            private RemindContent GenerateVehicleRemind(RoadLevelLimitedData data, MonChaosuSet setting, GPSPosition gps, String msgType, int limited) {
                try {
                    if (msgType.equalsIgnoreCase("预警短信")) {
                        RemindContent content = new RemindContent() {
                            {
                                VehicleId = gps.getVehId();
                                Ipaddress = gps.getTerminalPhoneNo();
                                VehicleNo = gps.getCheliangpaizhao();
                                PlateColor = gps.getChepaiyanse();
                                RemindNote = setting.getYjtxnr();
                                RemindType = 0x01;
                                RemindTime = new Date();
                                MessageType = "超速报警(道路)";
                                AlarmNote = "超速报警(道路)";
                                AlarmGuid = data.SingleGuid;
                                GpsTime = data.GpsTime;
                                LastSpeed = gps.getSpeedExt();
                            }


                        };
                        content.RemindNote = content.RemindNote.replace("{限速值}", limited + "");
                        return content;
                    } else if (msgType.equalsIgnoreCase("报警短信")) {
                        RemindContent content = new RemindContent() {
                            {
                                VehicleId = gps.getVehId();
                                Ipaddress = gps.getTerminalPhoneNo();
                                VehicleNo = gps.getCheliangpaizhao();
                                PlateColor = gps.getChepaiyanse();
                                RemindNote = setting.getCstxnr();
                                RemindType = 0x02;
                                RemindTime = new Date();
                                MessageType = "超速报警(道路)";
                                AlarmNote = "超速报警(道路)";
                                AlarmGuid = data.SingleGuid;
                                GpsTime = data.GpsTime;
                                LastSpeed = gps.getSpeedExt();
                            }

                        };
                        content.RemindNote = content.RemindNote.replace("{限速值}", limited + "");
                        return content;
                    } else {
                        return null;
                    }
                } catch (Exception ex) {
                    ex.printStackTrace();
                    return null;
                }
            }

            private Kafka_VehicleRoadLevelAlarm SendAlarmLevelToKafka(RoadLevelLimitedData data, GPSPosition gps, boolean isWarning) {

                Kafka_VehicleRoadLevelAlarm alarm = new Kafka_VehicleRoadLevelAlarm();
                alarm.setVehId(gps.getVehId());
                alarm.setDeptId(gps.getDeptId());
                alarm.setVehNo(gps.getCheliangpaizhao());
                alarm.setPlateColor(gps.getChepaiyanse());
                alarm.setAlarmFlag(1);
                alarm.setTime(gps.getTIME());
                alarm.setNote("超速报警(道路)");
                alarm.setLongitude(gps.getLngExt());
                alarm.setLatitude(gps.getLatExt());
                alarm.setLastlongitude(gps.getLngExt());
                alarm.setLastlatitude(gps.getLatExt());
                alarm.setVelocity(gps.getSpeedExt());
                alarm.setAngle(gps.getDIRECTION());
                alarm.setLocal(IntUtils.LocalExt(gps.getSTATUSBIT()) == 1 ? 1 : 0);
                alarm.setLimited(0);
                alarm.setAlarmBeginTime(data.getAlarmBeginTime());
                alarm.setAlarmEndTime(data.getAlarmEndTime());
                alarm.setAlarmTimeSpan(TimeUtils.getMinusSecond(data.getAlarmEndTime(), data.getAlarmBeginTime()));
                alarm.setMaxSpeed(data.getAlarmMaxSpeed());
                alarm.setDistance(data.getAlarmDistance() * 1000);
                alarm.setRoadLevel(data.getAlarmRoadLevel());
                alarm.setRoadLimited(data.getAlarmRoadMaxSpeed());
                alarm.setRoadName(data.getAlarmRoadName());
                alarm.setAlarmGuid(data.getSingleGuid());
                alarm.setAlarmLevel(data.getAlarmLevel());
                alarm.setKafkaFlag("Kafka_RoadLevelAlarmLevel");

                if (isWarning) {
                    alarm.setAlarmLevel(99);
                }
                return alarm;
            }

            private Kafka_VehicleRemindContent SendRemindToKafka(RemindContent remindContent) {
                Kafka_VehicleRemindContent kafkaVehicleRemindContent = new Kafka_VehicleRemindContent();
                kafkaVehicleRemindContent.setVehId(remindContent.getVehicleId());
                kafkaVehicleRemindContent.setVehNo(remindContent.getVehicleNo());
                kafkaVehicleRemindContent.setPlateColor(remindContent.getPlateColor());
                kafkaVehicleRemindContent.setRemindNote(remindContent.getRemindNote());
                kafkaVehicleRemindContent.setRemindTime(new Date());
                kafkaVehicleRemindContent.setRemindType(remindContent.getRemindType());
                kafkaVehicleRemindContent.setMessageType(remindContent.getMessageType());
                kafkaVehicleRemindContent.setAlarmNote(remindContent.getAlarmNote());
                kafkaVehicleRemindContent.setAlarmGuid(remindContent.getAlarmGuid());
                kafkaVehicleRemindContent.setGpsTime(remindContent.getGpsTime());
                kafkaVehicleRemindContent.setLastSpeed(remindContent.getLastSpeed());

                return kafkaVehicleRemindContent;


            }

            private RoadLevelResult DisposeMessage(RoadLevelLimitedData roadLevelLimitedData, MonChaosuSet mySetting, GPSPosition gps, String msgType, int limited, MonYanzhongweiguiSet severityAlarmSetting) {
                if (msgType.equalsIgnoreCase("预警短信")) {
                    if (mySetting.getYjenable() == 1 && mySetting.getYjtxcs() > 0 && StringUtils.isNotEmpty(mySetting.getYjtxnr())) {

                        if (severityAlarmSetting != null) {
                            if (severityAlarmSetting.getDingzhiguize() == 3) {
                                //新疆阿勒泰规则，只发一次预警，间隔时间内。
                                Long ts = TimeUtils.getMinusSecond(new Date(), roadLevelLimitedData.WarningContentSendTime_alt);
                                if (ts >= mySetting.getYjtxjg()) {
                                    roadLevelLimitedData.WarningContentSendTime_alt = new Date();
                                    RemindContent remindContent = GenerateVehicleRemind(roadLevelLimitedData, mySetting, gps, msgType, limited);

                                    return new RoadLevelResult(SendAlarmLevelToKafka(roadLevelLimitedData, gps, true), SendRemindToKafka(remindContent));
                                }
                            } else {
                                Long ts = TimeUtils.getMinusSecond(new Date(), roadLevelLimitedData.WarningContentSendTime);
                                if (ts >= mySetting.getYjtxjg() && roadLevelLimitedData.WarningContentCount < mySetting.getYjtxcs()) {
                                    roadLevelLimitedData.WarningContentCount++;
                                    roadLevelLimitedData.WarningContentSendTime = new Date();
                                    RemindContent remindContent = GenerateVehicleRemind(roadLevelLimitedData, mySetting, gps, msgType, limited);
                                    return new RoadLevelResult(SendAlarmLevelToKafka(roadLevelLimitedData, gps, true), SendRemindToKafka(remindContent));
                                }
                            }
                        } else {
                            Long ts = TimeUtils.getMinusSecond(new Date(), roadLevelLimitedData.WarningContentSendTime);
                            if (ts >= mySetting.getYjtxjg() && roadLevelLimitedData.WarningContentCount < mySetting.getYjtxcs()) {
                                roadLevelLimitedData.WarningContentCount++;
                                roadLevelLimitedData.WarningContentSendTime = new Date();
                                RemindContent remindContent = GenerateVehicleRemind(roadLevelLimitedData, mySetting, gps, msgType, limited);
                                return new RoadLevelResult(SendAlarmLevelToKafka(roadLevelLimitedData, gps, true), SendRemindToKafka(remindContent));
                            }
                        }
                    }
                } else if (msgType.equalsIgnoreCase("报警短信")) {

                    if (severityAlarmSetting != null) {
                        if (severityAlarmSetting.getQy1jftts() == 1 && roadLevelLimitedData.AlarmLevel < 1) {
                            //如果启用了大于1级才发短信规则，则退出，由分级那里处理短信
                            return null;
                        }
                    }
                    if (mySetting.getCstxcs() > 0 && StringUtils.isNotEmpty(mySetting.getCstxnr())) {
                        Long ts = TimeUtils.getMinusSecond(new Date(), roadLevelLimitedData.AlarmContentSendTime);
                        if (ts >= mySetting.getCstxjg() && roadLevelLimitedData.AlarmContentCount < mySetting.getCstxcs()) {
                            roadLevelLimitedData.AlarmContentCount++;
                            roadLevelLimitedData.AlarmContentSendTime = new Date();
                            RemindContent remindContent = GenerateVehicleRemind(roadLevelLimitedData, mySetting, gps, msgType, limited);
                            return new RoadLevelResult(null, SendRemindToKafka(remindContent));
                        }
                    }
                }
                return null;
            }

            private RoadLevelResult DisposeAlarmLevel(RoadLevelLimitedData roadLevelLimitedData, MonChaosuSet mySetting, GPSPosition gps, int limited, MonYanzhongweiguiSet severityAlarmSetting) {
                Kafka_VehicleRoadLevelAlarm kafkaVehicleRoadLevelAlarm = null;
                Kafka_VehicleRemindContent kafkaVehicleRemindContent = null;
                if (severityAlarmSetting != null) {
                    if (roadLevelLimitedData.AlarmRoadMaxSpeed == 0) return null;
                    if (roadLevelLimitedData.AlarmMaxSpeed == 0) return null;
                    double interval = TimeUtils.getMinusSecond(roadLevelLimitedData.AlarmEndTime, roadLevelLimitedData.AlarmBeginTime);
                    double rate = (Double.valueOf(roadLevelLimitedData.AlarmMaxSpeed - roadLevelLimitedData.AlarmRoadMaxSpeed) / roadLevelLimitedData.AlarmRoadMaxSpeed) * 100;
                    if (severityAlarmSetting.getQy5jgz() == 1) {
                        roadLevelLimitedData.AlarmSetting = 2;
                        // 新疆5级规则
                        int fifthDuration = 60;
                        int fifthRate = 50;

                        int fourthDuration = 60;
                        int fourthRate = 20;

                        int thirdDuration = 120;
                        int thirdRate = 10;

                        int secondDuration = 60;
                        int secondRate = 10;

                        int firstTtsDuration = 30;
                        int firstTtsRate = 10;
                        if (interval > fifthDuration && rate > fifthRate) {
                            //五级报警
                            if (roadLevelLimitedData.AlarmLevel < 5) {
                                roadLevelLimitedData.AlarmLevel = 5;
                                kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                            }
                            if (severityAlarmSetting.getQy1jftts() == 1) {
                                RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                if (tuple2 != null) {

                                    if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                        kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                    }

                                    if (tuple2.getKafkaVehicleRemindContent() != null) {
                                        kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                    }


                                }

                            }
                        } else if (interval > fourthDuration && rate > fourthRate) {
                            //四级报警
                            if (roadLevelLimitedData.AlarmLevel < 4) {
                                roadLevelLimitedData.AlarmLevel = 4;
                                kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                            }
                            if (severityAlarmSetting.getQy1jftts() == 1) {
                                RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                if (tuple2 != null) {

                                    if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                        kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                    }

                                    if (tuple2.getKafkaVehicleRemindContent() != null) {
                                        kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                    }


                                }
                            }
                        } else if (interval > thirdDuration && rate > thirdRate) {
                            //三级报警
                            if (roadLevelLimitedData.AlarmLevel < 3) {
                                roadLevelLimitedData.AlarmLevel = 3;
                                kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                            }
                            if (severityAlarmSetting.getQy1jftts() == 1) {
                                RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                if (tuple2 != null) {

                                    if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                        kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                    }

                                    if (tuple2.getKafkaVehicleRemindContent() != null) {
                                        kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                    }


                                }
                            }
                        } else if (interval > secondDuration && rate > secondRate) {
                            //二级报警
                            if (roadLevelLimitedData.AlarmLevel < 2) {
                                roadLevelLimitedData.AlarmLevel = 2;
                                kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                            }
                            if (severityAlarmSetting.getQy1jftts() == 1) {
                                RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                if (tuple2 != null) {

                                    if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                        kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                    }

                                    if (tuple2.getKafkaVehicleRemindContent() != null) {
                                        kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                    }


                                }
                            }
                        } else if (interval > firstTtsDuration && rate > firstTtsRate) {
                            if (roadLevelLimitedData.AlarmLevel < 1) {
                                roadLevelLimitedData.AlarmLevel = 1;
                                kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                            }
                            if (severityAlarmSetting.getQy1jftts() == 1) {
                                RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                if (tuple2 != null) {

                                    if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                        kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                    }

                                    if (tuple2.getKafkaVehicleRemindContent() != null) {
                                        kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                    }


                                }
                            }
                        }
                    } else if (severityAlarmSetting.getDingzhiguize() == 1) {
                        roadLevelLimitedData.AlarmSetting = 3;
                        // 新疆出租车规则
                        if (limited <= 40) {
                            int firstDuration = 90;
                            int firstRate = 20;

                            if (roadLevelLimitedData.AlarmMaxSpeed >= 60) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 1) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 1;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 40 && limited <= 50) {
                            int firstDuration = 90;
                            int firstRate = 10;
                            if (roadLevelLimitedData.AlarmMaxSpeed >= 60) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 1) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 1;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 50 && limited <= 60) {
                            int firstDuration = 90;
                            int firstRate = 20;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 60 && limited <= 70) {
                            int firstDuration = 90;
                            int firstRate = 10;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 70 && limited <= 80) {
                            int firstDuration = 90;
                            int firstRate = 10;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 80 && limited <= 90) {
                            int firstDuration = 90;
                            int firstRate = 10;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 90 && limited <= 120) {
                            int firstDuration = 90;
                            int firstRate = 5;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        }
                    } else if (severityAlarmSetting.getDingzhiguize() == 2) {
                        roadLevelLimitedData.AlarmSetting = 4;
                        // 新疆营运线路车

                        if (limited < 60) {
                            int firstDuration = 90;
                            int firstRate = 20;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 1) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 1;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited == 60) {
                            int secondDuration = 90;
                            int secondRate = 20;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited > 60 && limited < 100) {
                            int secondDuration = 90;
                            int secondRate = 10;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited >= 100 && limited <= 120) {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        }

                    } else if (severityAlarmSetting.getDingzhiguize() == 3) {
                        roadLevelLimitedData.AlarmSetting = 5;

                        //新疆阿勒泰规则

                        if (limited <= 40) {
                            int firstDuration = 90;
                            int firstRate = 20;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 1) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 1;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {

                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }

                                }
                            }

                            int times = 90;
                            if (roadLevelLimitedData.AlarmMaxSpeed >= 60 && interval >= times) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited <= 50) {
                            int firstDuration = 90;
                            int firstRate = 10;
                            if (interval >= firstDuration && rate >= firstRate) {
                                if (roadLevelLimitedData.AlarmLevel < 1) {
                                    //一级报警
                                    roadLevelLimitedData.AlarmLevel = 1;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }

                            int times = 90;
                            if (roadLevelLimitedData.AlarmMaxSpeed >= 60 && interval >= times) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            }
                        } else if (limited <= 60) {
                            int secondDuration = 90;
                            int secondRate = 20;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 70) {
                            int secondDuration = 90;
                            int secondRate = 20;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 80) {
                            int secondDuration = 90;
                            int secondRate = 10;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 90) {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 100) {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 110) {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else if (limited <= 120) {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        } else {
                            int secondDuration = 90;
                            int secondRate = 5;
                            if (interval >= secondDuration && rate >= secondRate) {
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    //二级报警
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                                if (severityAlarmSetting.getQy1jftts() == 1) {
                                    RoadLevelResult tuple2 = DisposeMessage(roadLevelLimitedData, mySetting, gps, "报警短信", limited, severityAlarmSetting);

                                    if (tuple2 != null) {

                                        if (tuple2.getKafkaVehicleRoadLevelAlarm() != null) {
                                            kafkaVehicleRoadLevelAlarm = tuple2.getKafkaVehicleRoadLevelAlarm();
                                        }

                                        if (tuple2.getKafkaVehicleRemindContent() != null) {
                                            kafkaVehicleRemindContent = tuple2.getKafkaVehicleRemindContent();
                                        }


                                    }
                                }
                            } else {

                            }
                        }
                    } else {
                        roadLevelLimitedData.AlarmSetting = 1;
                        // 通用规则
                        if (severityAlarmSetting.getCsbfb2() == 0) return null;
                        if (severityAlarmSetting.getCscxsj2() == 0) return null;
                        if (severityAlarmSetting.getCsbfb() == 0) return null;
                        if (severityAlarmSetting.getCscxsj() == 0) return null;
                        if (severityAlarmSetting.getCsbfb1() == 0) return null;
                        if (severityAlarmSetting.getCscxsj1() == 0) return null;

                        int thirdDuration = severityAlarmSetting.getCscxsj();
                        double thirdRate = severityAlarmSetting.getCsbfb();
                        int secondDuration = severityAlarmSetting.getCscxsj2();
                        double secondRate = severityAlarmSetting.getCsbfb2();
                        int firstDuration = severityAlarmSetting.getCscxsj1();
                        double firstRate = severityAlarmSetting.getCsbfb1();

                        if (severityAlarmSetting.getCspdgz() == 0) {
                            if (interval >= thirdDuration || rate >= thirdRate) {
                                //三级报警
                                if (roadLevelLimitedData.AlarmLevel < 3) {
                                    roadLevelLimitedData.AlarmLevel = 3;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                            } else if ((interval < secondDuration && interval >= firstDuration)
                                    || (rate < secondRate && rate >= firstRate)) {

                                //二级报警
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                            }

                        } else {
                            if (interval >= thirdDuration && rate >= thirdRate) {
                                //三级报警
                                if (roadLevelLimitedData.AlarmLevel < 3) {
                                    roadLevelLimitedData.AlarmLevel = 3;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                            } else if ((interval < secondDuration && interval >= firstDuration)
                                    && (rate < secondRate && rate >= firstRate)) {

                                //二级报警
                                if (roadLevelLimitedData.AlarmLevel < 2) {
                                    roadLevelLimitedData.AlarmLevel = 2;
                                    kafkaVehicleRoadLevelAlarm = SendAlarmLevelToKafka(roadLevelLimitedData, gps, false);
                                }
                            }

                        }
                    }


                }

                return new RoadLevelResult(kafkaVehicleRoadLevelAlarm, kafkaVehicleRemindContent);
            }


            @Override
            public RoadLevelResult map(Tuple3<GPSPosition, RoadLevelDataResult, MonChaosuSet> tuple3) throws Exception {
                GPSPosition gpsPosition = tuple3.f0;
                RoadLevelDataResult resultRoadLevelData = tuple3.f1;
                MonChaosuSet mySetting = tuple3.f2;
                RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(factory);
                RoadLevelResult roadLevelResult = new RoadLevelResult();
                SqlSession sqlSession = sqlSessionFactory.openSession(true);
                try {
                    MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = sqlSession.getMapper(MonYanzhongweiguiSetMapper.class);
                    TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);
                    MonYejianxiansuSetMapper monYejianxiansuSetMapper = sqlSession.getMapper(MonYejianxiansuSetMapper.class);


                    RoadLevelLimitedData roadLevelLimitedData = TryGetData_RLLD(gpsPosition.getVehId(), clusterRedisTemplate);

                    if (resultRoadLevelData == null) return null;
                    Pair<Boolean, Integer> p2 = TimeUtils.CheckData(roadLevelLimitedData.getGpsTime(), gpsPosition.TIME);
                    if (!p2.getKey()) {
                        return null;
                    } else {
                        roadLevelLimitedData.LocalTime = new Date();
                        roadLevelLimitedData.GpsTime = TimeUtils.toDate(gpsPosition.TIME);
                    }

                    int i = p2.getValue();
                    int limited = resultRoadLevelData.RoadMaxSpeed;
                    int lastMaxSpeed = roadLevelLimitedData.LastMaxSpeed;
                    //跟上一个真实渠道得限速值取大
                    limited = Math.max(limited, lastMaxSpeed);
                    //将当前真实限速值记录下来
                    roadLevelLimitedData.LastMaxSpeed = resultRoadLevelData.RoadMaxSpeed;
                    int ruleType = -1;
                    Pair<Boolean, MonYanzhongweiguiSet> p = GetSeverityRuleSetting(gpsPosition, monYanzhongweiguiSetMapper, tsAlarmRuleMapper);
                    Pair<Integer, Integer> p1 = CalcMaxSpeed(gpsPosition, p, limited);
                    int calclimited = p1.getKey();
                    ruleType = p1.getValue();

                    {
                        //计算是否在夜间限速
                        boolean inTimeRange = false;
                        MonYejianxiansuSet nightLimitedSetting = getVehicleForNightLimitedParamsDic(gpsPosition.getVehId(), monYejianxiansuSetMapper);
                        if (nightLimitedSetting != null) {
                            if (nightLimitedSetting.getEnable() == 1) {
                                inTimeRange = TimeUtils.InTimeRange(gpsPosition.TIME, nightLimitedSetting.getYjkssj(), nightLimitedSetting.getYjjssj());
                                if (inTimeRange) {
                                    limited = limited * Double.valueOf(nightLimitedSetting.getYjxsbl() / 100).intValue();
                                }
                            }
                        }
                    }

                    roadLevelLimitedData.setAlarmRoadMaxSpeed(limited);

                    Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);

                    if (mySetting != null) {

                        if (mySetting.getEnable() == 1) {
                            if (!roadLevelLimitedData.isIsCalc()) {
                                roadLevelLimitedData.ResetData();
                                roadLevelLimitedData.StartCalc(gpsPosition);
                            }
                            roadLevelLimitedData.WriteCalc(gpsPosition);

                            if (gpsPosition.getSpeedExt() > calclimited) {

                                if (!roadLevelLimitedData.isIsAlarm()) {

                                    roadLevelLimitedData.StartAlarm(gpsPosition, limited, resultRoadLevelData.getRoadLevel(), resultRoadLevelData.getRoadNameByLevel());
                                    //从没报警切换到报警，发送开始报警
                                    roadLevelResult.setKafkaVehicleRoadLevelAlarm(SendAlarmToKafka(gpsPosition, roadLevelLimitedData, true, driverLoginData));

                                }

                                //其他的时候记录报警的过程数据
                                roadLevelLimitedData.WriteAlarm(gpsPosition);
                                MonYanzhongweiguiSet monYanzhongweiguiSet = p.getValue();
                                RoadLevelResult rs = DisposeMessage(roadLevelLimitedData, mySetting, gpsPosition, "报警短信", limited, monYanzhongweiguiSet);
                                if (rs != null) {
                                    roadLevelResult.setKafkaVehicleRoadLevelAlarm(rs.getKafkaVehicleRoadLevelAlarm());
                                    roadLevelResult.setKafkaVehicleRemindContent(rs.getKafkaVehicleRemindContent());

                                }


                                RoadLevelResult resultLevel = DisposeAlarmLevel(roadLevelLimitedData, mySetting, gpsPosition, limited, p.getValue());


                                if (resultLevel != null) {
                                    if (rs != null) {
                                        if (rs.getKafkaVehicleRoadLevelAlarm() == null) {
                                            roadLevelResult.setKafkaVehicleRoadLevelAlarm(resultLevel.getKafkaVehicleRoadLevelAlarm());
                                        }
                                        if (rs.getKafkaVehicleRemindContent() == null) {
                                            roadLevelResult.setKafkaVehicleRemindContent(resultLevel.getKafkaVehicleRemindContent());
                                        }
                                    }


                                }

                            } else {
                                //速度小于限速值，停止报警
                                if (roadLevelLimitedData.IsAlarm) {
                                    //从报警状态切换未报警状态
                                    roadLevelLimitedData.StopAlarm();
                                    //从报警切换到没报警，发送停止报警
                                    roadLevelResult.setKafkaVehicleRoadLevelAlarm(SendAlarmToKafka(gpsPosition, roadLevelLimitedData, false, driverLoginData));
                                    roadLevelLimitedData.GenerateSingleGuid();
                                    roadLevelLimitedData.ResetData();
                                } else {
                                    //生成预警短信,如果生成过，则添加无效
                                    boolean isWaring = gpsPosition.getSpeedExt() > limited - mySetting.getYjzdsd();
                                    if (isWaring) {
                                        RoadLevelResult rs = DisposeMessage(roadLevelLimitedData, mySetting, gpsPosition, "预警短信", limited, p.getValue());
                                        if (rs != null) {
                                            roadLevelResult.setKafkaVehicleRoadLevelAlarm(rs.getKafkaVehicleRoadLevelAlarm());
                                            roadLevelResult.setKafkaVehicleRemindContent(rs.getKafkaVehicleRemindContent());
                                        }

                                    } else {
                                        roadLevelLimitedData.GenerateSingleGuid();
                                        roadLevelLimitedData.ResetData();
                                    }
                                }
                            }

                        } else {
                            //关闭了规则
                            if (roadLevelLimitedData.IsAlarm) {
                                //从报警状态切换未报警状态
                                roadLevelLimitedData.StopAlarm();
                                //从报警切换到没报警，发送停止报警

                                roadLevelResult.setKafkaVehicleRoadLevelAlarm(SendAlarmToKafka(gpsPosition, roadLevelLimitedData, false, driverLoginData));
                            }
                            roadLevelLimitedData.GenerateSingleGuid();
                            roadLevelLimitedData.ResetData();
                        }


                    } else {
                        //找不到规则
                        if (roadLevelLimitedData.IsAlarm) {
                            //从报警状态切换未报警状态
                            roadLevelLimitedData.StopAlarm();
                            //从报警切换到没报警，发送停止报警

                            roadLevelResult.setKafkaVehicleRoadLevelAlarm(SendAlarmToKafka(gpsPosition, roadLevelLimitedData, false, driverLoginData));
                        }
                        roadLevelLimitedData.GenerateSingleGuid();
                        roadLevelLimitedData.ResetData();
                    }


                    clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_RoadLevelLimited + gpsPosition.getVehId(), roadLevelLimitedData, 8, TimeUnit.HOURS);


                } catch (Exception e) {
                    e.printStackTrace();
                    sqlSession.rollback();
                } finally {
                    sqlSession.close();
                    return roadLevelResult;
                }
            }
        });
        SingleOutputStreamOperator<Kafka_VehicleRoadLevelAlarm> kafkaVehicleRoadLevelAlarmStream= roadLevelResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleRoadLevelAlarm() != null)
                .map(e -> e.getKafkaVehicleRoadLevelAlarm());

       SingleOutputStreamOperator<KafkaProcessResult> tuple3SingleOutputStreamOperator= kafkaVehicleRoadLevelAlarmStream
                .filter(e->e.getKafkaFlag().equalsIgnoreCase("Kafka_RoadLevelAlarm"))
                .map(new KafkaRoadLevelAlarmProcess())
                .filter(Objects::nonNull);

        tuple3SingleOutputStreamOperator
                .filter(e->e.getMonitorClientAlarm()!=null)
                .map(e->e.getMonitorClientAlarm())
                .print();

        tuple3SingleOutputStreamOperator
                .filter(e->e.getBaobiaoAlarmsummary()!=null)
                .map(e->e.getBaobiaoAlarmsummary())
                .addSink(new BaobiaoAlarmsummarySink());

        tuple3SingleOutputStreamOperator
                .filter(e->e.getBaobiaoAlarmsummaryRecord()!=null)
                .map(e->e.getBaobiaoAlarmsummaryRecord())
                .addSink(new BaobiaoAlarmsummaryRecordSink());

        SingleOutputStreamOperator<KafkaProcessResult>  levelStream= kafkaVehicleRoadLevelAlarmStream
                .filter(e->e.getKafkaFlag().equalsIgnoreCase("Kafka_RoadLevelAlarmLevel"))
                .map(new KafkaRoadLevelAlarmLevelProcess())
                .filter(Objects::nonNull);

        levelStream
                .filter(e->e.getMonitorClientAlarm()!=null)
                .map(e->e.getMonitorClientAlarm())
                .print();

        levelStream
                .filter(e->e.getBaobiaoAlarmsummary()!=null)
                .map(e->e.getBaobiaoAlarmsummary())
                .addSink(new BaobiaoAlarmsummarySink());

        levelStream
                .filter(e->e.getBaobiaoAlarmsummaryRecord()!=null)
                .map(e->e.getBaobiaoAlarmsummaryRecord())
                .addSink(new BaobiaoAlarmsummaryRecordSink());


        roadLevelResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleRemindContent() != null)
                .map(e->e.getKafkaVehicleRemindContent())
                .print();
    }
}
